package com.atguigu.day08;

import com.atguigu.beans.WaterSensor;
import com.atguigu.func.WaterSensorMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/4/9
 * 该案例演示了键控状态-值状态
 * 需求：检测每种传感器的水位值，如果连续的两个水位差值超过10，就输出报警
 */
public class Flink02_KeyedState_ValueState_2 {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);
        //TODO 2.从指定的网络端口读取数据
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        //TODO 3.对读取的数据进行类型转换   String->WaterSensor
        SingleOutputStreamOperator<WaterSensor> wsDS = socketDS.map(new WaterSensorMapFunction());
        //TODO 4.按照传感器id进行分组
        KeyedStream<WaterSensor, String> keyedDS = wsDS.keyBy(WaterSensor::getId);
        //TODO 5.对分组后的数据进行处理
        SingleOutputStreamOperator<String> processDS = keyedDS.process(
                new KeyedProcessFunction<String, WaterSensor, String>() {
                    //声明一个值状态，用于存放上次水位值
                    //键控状态和普通的成员变量区别
                    //1.作用范围不同   普通的成员变量区别作用于整个并行度   键控状态作用于keyby后的每一个组
                    //2.普通的成员变量存储于内存，不能被持久化 ；键控状态可以被持久化
                    //注意：不能在声明的时候直接对状态进行初始化， 这个时候还获取不到运行时上下文对象
                    //ValueState<Integer> lastVcState = getRuntimeContext().getState(new ValueStateDescriptor<>("lastVcState", Integer.class));
                    ValueState<Integer> lastVcState;

                    //给状态初始化
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<Integer> valueStateDescriptor
                                = new ValueStateDescriptor<Integer>("lastVcState", Integer.class);
                        lastVcState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(WaterSensor ws, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                        //获取上次水位值
                        Integer lastVc = lastVcState.value() == null ?0 :  lastVcState.value();
                        //获取当前水位值
                        Integer curVc = ws.getVc();
                        //获取传感器id
                        String id = ctx.getCurrentKey();
                        if (Math.abs(curVc - lastVc) > 10) {
                            out.collect("传感器" + id + "的当前水位值" + curVc + "和上一次水位值" + lastVc + "之差大于10");
                        }
                        //将当前水位值更新到状态中
                        lastVcState.update(curVc);
                    }
                }
        );

        //TODO 6.打印输出
        processDS.printToErr();
        //TODO 7.提交作业
        env.execute();
    }
}
